package com.rabbitmq.origin.api;

import com.rabbitmq.client.*;
import com.rabbitmq.common.utils.RabbitMqUtil;
import lombok.extern.slf4j.Slf4j;

import java.io.IOException;

/**
 * @author
 * @Describe
 * @date
 */
@Slf4j(topic = "RabbitMqConsumerExchange2Queue")
public class RabbitMqConsumerExchange2Queue {

    /**
     * 消费者
     *
     * @param exchangeName
     * @param exchangeType
     */
    public void consumerExchange2Queue(String exchangeName, String exchangeType,
                                       String routingKey, String queueName, int flag) {
        Connection conn = null;
        Channel channel = null;
        try {
            // 获取连接
            conn = RabbitMqUtil.getConn();
            // 创建通道（通道可用于消息的发送和接收）
            channel = conn.createChannel();

            // 声明交换机
            channel.exchangeDeclare(exchangeName, exchangeType, true);

            // 声明队列，队列已存在不会重新创建
            channel.queueDeclare(queueName, true, false, false, null);

            // 绑定队列
            channel.queueBind(queueName, exchangeName, routingKey);

            // 创建队列消费者
            final Consumer consumer = new DefaultConsumer(channel) {
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties,
                                           byte[] body) throws IOException {
                    String message = new String(body, "UTF-8");
                    System.out.println(consumerTag + " consumer-" + flag + " handleDelivery " + envelope.getRoutingKey() + ":'" + message + "'");
                }
            };

            // 消息消费，自动确认
            channel.basicConsume(queueName, true, exchangeName + " & " + queueName, consumer);
        } catch (Exception e) {
            log.error("消费消息异常,", e);
            // 资源关闭
            RabbitMqUtil.close(conn, channel);
        }
    }
}
